package operator.filter;

import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import util.Printer;
import util.RxJavaHelper;
import util.ThreadHelper;

class DebounceOperator {

    public static void main(String[] args) {
//        debounce1();
        debounce2();
    }

    /**
     * 限流操作符，过滤掉发射速率过快的数据项
     * 在一个数据发射后的一个时间段内，若没有新数据发射时，发射该数据；
     * 若有，则在新数据到来时重新计时，重复之前所叙逻辑
     * 但是不会限制终止操作符（onError或onCompleted）
     */
    static void debounce1(){
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                // send events with simulated time wait
                emitter.onNext(1); // skip
                Thread.sleep(400);
                emitter.onNext(2); // deliver
                Thread.sleep(505);
                emitter.onNext(3); // skip
                Thread.sleep(100);
                emitter.onNext(4); // deliver
                Thread.sleep(605);
                emitter.onNext(5); // deliver
                Thread.sleep(510);
                emitter.onComplete();
            }
        })
                .debounce(500, TimeUnit.MILLISECONDS)
                .subscribe(RxJavaHelper.getConsumer());
        ThreadHelper.sleep();
    }


    /**
     *原始Observable的每一项应用一个函数进行限流，这个函数返回一个Observable。
     * 如果原始Observable在这个新生成的Observable终止之前发射了另一个数据，
     * debounce会抑制(suppress)这个数据项。
     */
    static void debounce2(){
        Observable.intervalRange(1,10, 0, 1, TimeUnit.SECONDS)
                .doOnNext(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Printer.print("---------------- doOnNext aLong = " + aLong);
                    }
                })
                .debounce(new Function<Long, ObservableSource<Long>>() {
                    @Override
                    public ObservableSource<Long> apply(Long aLong) throws Exception {
                        if(aLong > 5)
                            return Observable.empty();
                        else
                            return Observable.never();
                    }
                })
                .subscribe(RxJavaHelper.getConsumer());
        ThreadHelper.sleep();
    }


}
